package cc.spider.core;

import cc.spider.exception.ConfigException;
import cc.spider.model.FlowData;
import cc.spider.model.ResultInfo;
import cc.spider.model.SpiderConfig;
import cc.spider.model.TaskInfo;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class SpiderCore {

    /**
     * 数据管道
     */
    protected List<IDataPipe> dataPipes = new ArrayList<>();

    /**
     * 爬虫配置
     */
    protected SpiderConfig config;


    private Lock stateLock = new ReentrantLock();
    private Condition stateCondition = stateLock.newCondition();

    /**
     * 设置爬虫状态 0是默认，1是运行，2是暂停，3是退出
     */
    private byte action = 0;

    /**
     * 标记action是否完成执行
     */
    private boolean actionComplete = true;

    // 线程现在的状态
    // 5个状态： 0、默认，1、已启动，2、运行中，3、已停止，4、已退出
    private volatile byte state = 0;
    private volatile boolean isRunning = false;
    private volatile boolean isStart = false;

    protected Lock stepLock = new ReentrantLock();
    protected Condition emptyQueue = stepLock.newCondition();

    private long waitingTimeout = 30;

    private Thread runThread = null;
    private volatile long stopSleep = 0;

    public SpiderCore(SpiderConfig spiderConfig) {
        this.config = spiderConfig;
        if (this.config.getFetcher() == null || this.config.getDataProcess() == null) {
            throw new ConfigException("配置错误");
        }
    }


    /**
     * 获取状态
     *
     * @return
     */
    public int getState() {
        return state;
    }

    /**
     * 获取爬虫运行线程
     *
     * @return
     */
    public Thread getRunThread() {
        return runThread;
    }

    /**
     * 启动爬虫
     * 只有线程未启动过或者线程已经退出了的情况下才会重新启动
     */
    public synchronized boolean start() {
        if (action == 0 || (state == 4 && !runThread.isAlive())) {
            actionComplete = false;
            action = 1;
            isStart = true;
            isRunning = true;
            runThread = new Thread(this::run);
            runThread.start();
            return true;
        } else {
            return false;
        }
    }

    /**
     * 停止（不会立即停止）
     *
     * @param sleep 暂停的时间，单位是毫秒。如果sleep为0则表示不限时间
     */
    public synchronized boolean stop(long sleep) {
        if (action == 1 && actionComplete) {
            actionComplete = false;
            action = 2;
            stopSleep = sleep;
            isRunning = false;
            return true;
        } else {
            return false;
        }
    }

    public boolean stop() {
        return this.stop(0);
    }

    /**
     * 从停止中恢复
     */
    public synchronized boolean resume() {
        if (action == 2 && actionComplete) {
            actionComplete = false;
            action = 1;
            isRunning = true;
            stateLock.lock();
            stateCondition.signalAll();
            stateLock.unlock();
            return true;
        } else {
            return false;
        }
    }

    /**
     * 退出（不会立即退出）
     */
    public synchronized boolean exit() {
        if (isStart && actionComplete) {
            actionComplete = false;
            action = 3;
            isStart = false;
            isRunning = false;
            stateLock.lock();
            stateCondition.signalAll();
            stateLock.unlock();
            return true;
        } else {
            return false;
        }
    }


    private void run() {
        stateListenerProxy.start();
        while (isStart) { // 控制退出的循环
            while (isRunning) { // 控制暂停的循环
                state = 2; // 设置成运行中
                this.execute();
            }
            if (isStart) {
                stateLock.lock();
                try {
                    stateListenerProxy.stop();
                    if (stopSleep > 0) {
                        stateCondition.await(stopSleep, TimeUnit.MILLISECONDS);
                        isRunning = true;
                        action = 1;
                    } else {
                        stateCondition.await();
                    }
                    stateListenerProxy.resume();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    stateLock.unlock();
                }
            }
        }
        stateListenerProxy.exit();
    }

    private void execute() {
        TaskInfo taskInfo = this.config.getTaskQueue().dequeue();
        if (taskInfo == null) {
            stepLock.lock();
            try {
                emptyQueue.await(waitingTimeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                stepLock.unlock();
            }
        } else {
            this.config.getThreadPool().addTask(() -> {
                fetch(taskInfo);
            });
            try {
                Thread.sleep(this.config.getSleepTime());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    protected IFetcherListener fetcherListener = new IFetcherListener() {
        @Override
        public void success(TaskInfo taskInfo, ResultInfo resultInfo) {
            FlowData flowData = new FlowData(taskInfo, resultInfo);
            try {
                // 处理数据
                process(flowData);
                // 将 task 加入队列
                addTasks(flowData.getNextTaskInfos());
                // 处理解析完的数据
                dataPipe(flowData);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void fail(TaskInfo taskInfo, ResultInfo resultInfo) {
            System.err.println("fetch fail");
        }
    };

    public void addTask(TaskInfo taskInfo) {
        if (taskInfo != null) {
            this.config.getTaskQueue().enqueue(taskInfo);
            try {
                stepLock.lock();
                emptyQueue.signalAll();
            } finally {
                stepLock.unlock();
            }
        }
    }

    public void addTasks(List<TaskInfo> taskInfos) {
        if (taskInfos != null && taskInfos.size() > 0) {
            for (TaskInfo taskInfo : taskInfos) {
                this.config.getTaskQueue().enqueue(taskInfo);
            }
            try {
                stepLock.lock();
                emptyQueue.signalAll();
            } finally {
                stepLock.unlock();
            }
        }
    }

    /**
     * 获取数据
     *
     * @param task
     */
    private void fetch(TaskInfo task) {
        IFetcher fetcher = getFetcher(task);
        fetcher.fetch(task, config.getGlobalDefaultData(), fetcherListener);
    }

    /**
     * 处理响应数据（获取需要的内容）
     *
     * @param flowData
     */
    private void process(FlowData flowData) {
        IDataProcessor spiderProcess = getSpiderProcess(flowData.getTaskInfo());
        spiderProcess.process(flowData);
    }

    /**
     * 持久化数据
     *
     * @param flowData
     */
    private void dataPipe(FlowData flowData) {
        List<IDataPipe> dataPipeList = getDataPipes(flowData.getTaskInfo());
        for (IDataPipe dataPipe : dataPipeList) {
            dataPipe.write(flowData);
        }
    }

    /**
     * 返回一个获取器。如果TaskInfo设置了fetcher就返回这个，否则返回默认的（config设置的)fetcher
     *
     * @param taskInfo
     * @return
     */
    protected IFetcher getFetcher(TaskInfo taskInfo) {
        IFetcher fetcher = taskInfo.getFetcher();
        return fetcher == null ? this.config.getFetcher() : fetcher;
    }

    /**
     * 返回一个处理器。如果TaskInfo设置了processor就返回这个，否则返回默认的（config设置的)processor
     *
     * @param taskInfo
     * @return
     */
    protected IDataProcessor getSpiderProcess(TaskInfo taskInfo) {
        IDataProcessor processor = taskInfo.getProcessor();
        return processor == null ? this.config.getDataProcess() : processor;
    }


    /**
     * 返回一个持久化器。如果TaskInfo设置了DataPipe就返回这个，否则返回默认的（config设置的)DataPipe
     *
     * @param taskInfo
     * @return
     */
    protected List<IDataPipe> getDataPipes(TaskInfo taskInfo) {
        List<IDataPipe> dataPipeList = taskInfo.getDataPipeList();
        return dataPipeList == null ? this.config.getDataPipeList() : dataPipeList;
    }

    public interface IStateListener {
        void start();

        void exit();

        void stop();

        void resume();

    }

    private IStateListener stateListener = new IStateListener() {
        @Override
        public void start() {
            System.out.println("启动成功");
        }

        @Override
        public void exit() {
            System.out.println("退出成功");
        }

        @Override
        public void stop() {
            System.out.println("暂停中。。。");
        }

        @Override
        public void resume() {
            System.out.println("从暂停中恢复");
        }
    };

    public void setStateListener(IStateListener stateListener) {
        this.stateListener = stateListener;
    }

    private IStateListener stateListenerProxy = new StateListenerProxy();

    private class StateListenerProxy implements IStateListener {
        @Override
        public void start() {
            state = 1; // 设置成已启动
            actionComplete = true;
            stateListener.start();

        }

        @Override
        public void exit() {
            state = 4; // 设置成已退出
            actionComplete = true;
            stateListener.exit();

        }

        @Override
        public void stop() {
            state = 3;// 设置成已停止
            actionComplete = true;
            stateListener.stop();

        }

        @Override
        public void resume() {
            state = 2; // 设置成运行中
            actionComplete = true;
            stateListener.resume();

        }
    }
}
